package cn._51doit.kafka.clients;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

public class JavaConsumer {

    public static void main(String[] args) {

        Properties properties = new Properties();
        //指定kafka broker的地址
        properties.setProperty("bootstrap.servers", "node-1.51doit.cn:9092,node-2.51doit.cn:9092,node-3.51doit.cn:9092");
        //指定写入数据key的反序列化方式
        properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        //指定写入数据value的反序列化方式
        properties.setProperty("value.deserializer", StringDeserializer.class.getName());
        //指定消费者组id（必须的）
        properties.setProperty("group.id", "g0001");
        //指定从什么位置开始读取数据
        properties.setProperty("auto.offset.reset","earliest");

        KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);

        //消费可以监听一到多个topic
        List<String> topics = Arrays.asList("test", "wordcount");

        kafkaConsumer.subscribe(topics);
//
//        ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(5));
//
//        for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
//            System.out.println(consumerRecord);
//        }

        while (true) {
            //poll拉取数据等待的时长
            ConsumerRecords<String, String> consumerRecords = kafkaConsumer.poll(Duration.ofSeconds(5));
            //一次会拉取一个批次（0到多条数据）
            for (ConsumerRecord<String, String> consumerRecord : consumerRecords) {
                System.out.println(consumerRecord);
            }
            System.out.println("88888888");
        }


    }
}
